﻿// This code is released under the BSD license.
namespace PugLib.IO
{
    // Stephen Toub is the original author
    using System;
    using System.Collections.Generic;
    using System.IO;
    using System.Threading;

    /// <summary>A Stream for passing data between producers and consumers.</summary>
    public class BlockingStream : Stream
    {
        /// <summary>Chunks of data written to the stream that have not yet been read.</summary>
        private readonly Queue<byte[]> _chunks;

        /// <summary>The index into _waitHandles for the _doneWriting event.</summary>
        private readonly int _doneWritingHandleIndex;

        /// <summary>Array of events waited on by readers.</summary>
        private readonly WaitHandle[] _events;

        /// <summary>Used as a monitor to protect state accessible to callers of both Read and Write.</summary>
        private readonly object _lockForAll;

        /// <summary>Used as a monitor to protect state accessible to callers of Read.</summary>
        private readonly object _lockForRead;

        /// <summary>The chunk currently being read from.</summary>
        private byte[] _currentChunk;

        /// <summary>The position in the current stream from which the next read request should be satisfied.</summary>
        private int _currentChunkPosition;

        /// <summary>Event used to signal that data is available for reading.</summary>
        private ManualResetEvent _dataAvailable;

        /// <summary>Event used to signal when no more data will be written to the stream.</summary>
        private ManualResetEvent _doneWriting;

        /// <summary>
        /// Flag used to determine whether write requests are invalid due to the stream already
        /// having been closed for writing.
        /// </summary>
        private bool _illegalToWrite;

        /// <summary>Initializes the blocking stream.</summary>
        public BlockingStream()
        {
            _chunks = new Queue<byte[]>();
            _doneWriting = new ManualResetEvent(false);
            _dataAvailable = new ManualResetEvent(false);
            _events = new WaitHandle[] { _dataAvailable, _doneWriting };
            _doneWritingHandleIndex = 1;
            _lockForRead = new object();
            _lockForAll = new object();
        }

        /// <summary>Determines whether data can be read from the stream.</summary>
        public override bool CanRead
        {
            get { return true; }
        }

        /// <summary>Determines whether the stream can be seeked.</summary>
        public override bool CanSeek
        {
            get { return false; }
        }

        /// <summary>Determines whether data can be written to the stream.</summary>
        public override bool CanWrite
        {
            get { return !_illegalToWrite; }
        }

        /// <summary>Not supported.</summary>
        public override long Length
        {
            get { throw new NotSupportedException(); }
        }

        /// <summary>Not supported.</summary>
        public override long Position
        {
            get { throw new NotSupportedException(); }
            set { throw new NotSupportedException(); }
        }

        /// <summary>Flush is not applicable to BlockingStream, but using it will not harm its operation.</summary>
        public override void Flush()
        {
        }

        /// <summary>Not supported.</summary>
        public override long Seek(long offset, SeekOrigin origin)
        {
            throw new NotSupportedException();
        }

        /// <summary>Not supported.</summary>
        public override void SetLength(long value)
        {
            throw new NotSupportedException();
        }

        /// <summary>
        /// Reads a sequence of bytes from the stream.  Read will block until at least one byte is available for
        /// reading or until the stream has been ended for writing.
        /// </summary>
        /// <param name="buffer">
        /// An array of bytes. When Read returns, the buffer contains the specified
        /// byte array with the values between offset and (offset + count - 1) replaced
        /// by the bytes read from the current source.
        /// </param>
        /// <param name="offset">
        /// The zero-based byte offset in buffer at which to begin storing the data read
        /// from the current stream.
        /// </param>
        /// <param name="count">The maximum number of bytes to be read from the current stream.</param>
        /// <returns>
        /// The total number of bytes read into the buffer. This can be less than the
        /// number of bytes requested if that many bytes are not currently available,
        /// or zero if the end of the stream has been reached.
        /// </returns>
        /// <remarks>This method is thread-safe.</remarks>
        public override int Read(byte[] buffer, int offset, int count)
        {
            // Validate all arguments and instance state
            if (buffer == null)
            {
                throw new ArgumentNullException("buffer");
            }
            if (offset < 0 || offset >= buffer.Length)
            {
                throw new ArgumentOutOfRangeException("offset");
            }
            if (count < 0 || offset + count > buffer.Length)
            {
                throw new ArgumentOutOfRangeException("count");
            }
            if (_dataAvailable == null)
            {
                throw new ObjectDisposedException(GetType().Name);
            }

            // If no data has been requested, we don't need to do anything
            if (count == 0)
            {
                return 0;
            }

            // Loop until we get data or until writing has completed
            while (true)
            {
                // Wait for either data to be available or for writing to be completed
                int handleIndex = WaitHandle.WaitAny(_events);

                lock (_lockForRead)
                {
                    lock (_lockForAll)
                    {
                        // If no chunk is currently selected, we need to get a chunk from the queue.  However,
                        // if there are no chunks, then loop back and wait again (the exception to this is that 
                        // if writing has been completed and there are no chunks, we're done).
                        if (_currentChunk == null)
                        {
                            if (_chunks.Count == 0)
                            {
                                if (handleIndex == _doneWritingHandleIndex)
                                {
                                    return 0;
                                }
                                else
                                {
                                    continue;
                                }
                            }
                            _currentChunk = _chunks.Dequeue();
                            _currentChunkPosition = 0;
                        }
                    }

                    // Determine how many bytes are available in the current chunk
                    int bytesAvailable = _currentChunk.Length - _currentChunkPosition;
                    int bytesToCopy;
                    if (bytesAvailable > count)
                    {
                        // Enough bytes are available to fill the whole request, so do so
                        bytesToCopy = count;
                        Buffer.BlockCopy(_currentChunk, _currentChunkPosition, buffer, offset, count);
                        _currentChunkPosition += count;
                    }
                    else
                    {
                        // Bytes are only available to fill part of the request.  Return those,
                        // and if there are no more chunks available in the queue, also reset
                        // the _dataAvailable event to continue to block waiting readers.
                        bytesToCopy = bytesAvailable;
                        Buffer.BlockCopy(_currentChunk, _currentChunkPosition, buffer, offset, bytesToCopy);
                        _currentChunk = null;
                        _currentChunkPosition = 0;
                        lock (_lockForAll)
                        {
                            if (_chunks.Count == 0)
                            {
                                _dataAvailable.Reset();
                            }
                        }
                    }
                    return bytesToCopy;
                }
            }
        }

        /// <summary>Writes a sequence of bytes to the stream.</summary>
        /// <param name="buffer">An array of bytes. Write copies count bytes from buffer to the stream.</param>
        /// <param name="offset">The zero-based byte offset in buffer at which to begin copying bytes to the stream.</param>
        /// <param name="count">The number of bytes to be written to the current stream.</param>
        /// <remarks>This method is thread-safe.</remarks>
        public override void Write(byte[] buffer, int offset, int count)
        {
            // Validate all arguments and instance state
            if (buffer == null)
            {
                throw new ArgumentNullException("buffer");
            }
            if (offset < 0 || offset >= buffer.Length)
            {
                throw new ArgumentOutOfRangeException("offset");
            }
            if (count < 0 || offset + count > buffer.Length)
            {
                throw new ArgumentOutOfRangeException("count");
            }
            if (_dataAvailable == null)
            {
                throw new ObjectDisposedException(GetType().Name);
            }

            // If no data is being written, nothing to do.
            if (count == 0)
            {
                return;
            }

            // Copy the written data into new byte array, and store that byte array into our queue of chunks.
            // Make sure to signal to any readers that data is available.
            byte[] chunk = new byte[count];
            Buffer.BlockCopy(buffer, offset, chunk, 0, count);
            lock (_lockForAll)
            {
                if (_illegalToWrite)
                {
                    throw new InvalidOperationException("Writing has already been completed.");
                }
                _chunks.Enqueue(chunk);
                _dataAvailable.Set();
            }
        }

        /// <summary>Signals to readers that no more data will be written to the stream.</summary>
        /// <remarks>This method is thread-safe.</remarks>
        public void SetEndOfStream()
        {
            if (_dataAvailable == null)
            {
                throw new ObjectDisposedException(GetType().Name);
            }
            lock (_lockForAll)
            {
                _illegalToWrite = true;
                _doneWriting.Set();
            }
        }

        /// <summary>Closes the stream and releases all resources associated with it.</summary>
        /// <remarks>This method is not thread-safe.</remarks>
        public override void Close()
        {
            base.Close();

            // Free the events
            if (_dataAvailable != null)
            {
                _dataAvailable.Close();
                _dataAvailable = null;
            }
            if (_doneWriting != null)
            {
                _doneWriting.Close();
                _doneWriting = null;
            }
        }
    }
}